package com.shujia.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo21Stage {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setAppName("pi")
      .setMaster("local")

    val sc = new SparkContext(conf)


    val linesRDD: RDD[String] = sc.textFile("spark/data/words")

    /**
      *
      * 默认一个block对应一个分区，每一个分区都会由一个task处理
      * 通过宽依赖切分Stage, 每一个stage是一组可以进行并行计算的task
      *
      *
      */


    //获取rddf分区数量
    println("linesRDD分区数:" + linesRDD.getNumPartitions)


    val wordsRDD: RDD[String] = linesRDD.flatMap(_.split(","))


    val kvRDD: RDD[(String, Int)] = wordsRDD.map((_, 1))

    println("kvRDD分区数量：" + kvRDD.getNumPartitions)

    //    kvRDD.reduceByKey((x, y) => x + y)
    /**
      *
      * shuffle之后rdd分区的数量默认等于shuffle之前rdd分区的数量
      *
      *
      *
      * shuffle类的算子可以手动指定分区的数量
      *
      */

    val countRDD: RDD[(String, Int)] = kvRDD.reduceByKey(_ + _, 4)


    println("countRDD分区数量：" + countRDD.getNumPartitions)

    countRDD.foreach(println)


    while (true) {

    }

  }

}
